package com.edgar.hbase;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.logging.Logger;


/**
 * @author 李沁声
 * @version 1.0
 * @date 2020/3/19 15:51
 * @Ues 连接到 Hbase 数据库，提供存储和查询功能
 */
public class HbaseClient {

    private Logger logger = Logger.getLogger(HbaseClient.class.toString());
    private Admin admin = null;
    private Connection connection = null;
    private Configuration configuration = null;
    private String ipAddress;

    public HbaseClient(String ipAddress, String port){
        this.ipAddress = ipAddress;

        // HBase配置文件  maxsize表示可上传文件大小变为 100 M
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", this.ipAddress);
        configuration.set("hbase.zookeeper.property.clientPort", port);
        configuration.set("hbase.client.keyvalue.maxsize","104857600");
        //获取连接对象
        try {
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 连接后关闭资源
    private  void close(Connection conn,Admin admin){
        if(conn != null){
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if(admin != null){
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    // 判断表存在
    public boolean tableExist(String tablename) {

        boolean tableExists = false;
        try {
            tableExists = admin.tableExists(TableName.valueOf(tablename));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return tableExists;
    }

    //    创建表(表名和列族) ---后面是可以设置多个字符串，即多个列族
    public void creatTable(String tableName, String...cfs)  {
        //判断表存在
        if(tableExist(tableName)){
            logger.info("表" + tableName + "存在");
            return;
        }
        //创建表描述器
        HTableDescriptor hTableDescriptor =  new HTableDescriptor(TableName.valueOf(tableName));
        //添加列族
        for(String cf:cfs){
            //创建。列描述器
            HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(cf);
            hTableDescriptor.addFamily(hColumnDescriptor);
        }
        //创建表操作
        try {
            admin.createTable(hTableDescriptor);
        } catch (IOException e) {
            e.printStackTrace();
        }
        logger.info("创建表" + tableName + "成功");
    }

    //    删除表
    public void deleteTable(String tableName)  {
        try{
            //判断表存在
            if(!tableExist(tableName)){
                return;
            }
            admin.disableTable(TableName.valueOf(tableName));
            admin.deleteTable(TableName.valueOf(tableName));
        }catch (IOException e){
        }
        logger.info("表" + tableName + "已删除");
    }

    // 增、改(表名、行名、列族、列名、数据)
    public void putData(String tableName, String rowKey, String cf, String cn, byte[] value){
        try {
            //获取白表对象
            Table table =  connection.getTable(TableName.valueOf(tableName));
            //创建put对象
            Put put = new Put(Bytes.toBytes(rowKey));
            //添加数据
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn), Bytes.toBytes(ByteBuffer.wrap(value)));
            //执行操作
            table.put(put);
            table.close();
        }catch (IOException e){
        }
        // 后期统一关闭  不需要进行关闭，因为每次打开 表格 的时候都关闭了
//        close(connection,admin);
    }

    //删除表里面的内容
    public void delete(String tableName, String rowKey, String cf, String cn){
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            //创建delete对象
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            //删除到指定的列和值
            delete.addColumns(Bytes.toBytes(cf),Bytes.toBytes(cn));
            table.delete(delete);
            table.close();
        }catch (IOException e){
        }
    }

    //查,全表扫描
    public void scanTable(String tableName){
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            //设置扫描范围
//            scan.setStartRow()
            ResultScanner results = table.getScanner(scan);
            //遍历数据打印(一个rowKey下多个列，多个版本)
            for(Result result:results ){
                Cell[] cells = result.rawCells();
                for(Cell cell:cells){
                    System.out.println("RK:" + Bytes.toString(CellUtil.cloneRow(cell))
                            +",CF:" + Bytes.toString(CellUtil.cloneFamily(cell) )
                            +",CN:" + Bytes.toString(CellUtil.cloneQualifier(cell))
                            +",VALUE:" + Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }
            table.close();
        }catch (IOException e){
        }
    }

    //获取指定列族，列的数据
    public byte[] getData(String tableName, String rowKey, String cf, String cn){
        byte[] Value = null;
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            // 创建get对象
            Get get = new Get(Bytes.toBytes(rowKey));
            //指定到列族或列
            get.addColumn(Bytes.toBytes(cf),Bytes.toBytes(cn));
            //获取最新的版本
            Result results = table.get(get);
            Cell[] cells = results.rawCells();
            for(Cell cell:cells){
                Value = CellUtil.cloneValue(cell);
            }
            table.close();
        }catch (IOException e){
        }
//        close(connection,admin);
        return Value;
    }

    // 关闭连接
    public void closeCollection(){
        close(connection,admin);
    }

}
